package org.midy.base;

import org.midy.infrastructure.tools.Logger;
import org.midy.infrastructure.tools.ThreadTools;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ClientLongConnectionPool {
    private final static LinkedList<BlockingDeque<ConnectClient>> taskList = new LinkedList<>();
    private static final Thread thread;

    static {
        thread = new Thread(() -> {
            try {
                final byte[] heartbeatBytes = "heart beat[心跳]".getBytes(StandardCharsets.UTF_8);
                int firstConnectionIdx = 0;
                while (true) {
                    Logger.info("开始心跳处理");
                    ConnectClient firstConnectClient = null;
                    try {
                        for (int idx = 0, location = firstConnectionIdx, max = taskList.size(); idx < max; idx++, location = (++location) % max) {
                            ConnectClient peekClient = null;
                            BlockingDeque<ConnectClient> queue = taskList.get(location);

                            if (queue == null) {
                                continue;
                            }

                            ConnectClient client = null;

                            try {

                                while ((peekClient = queue.peek()) != null && (peekClient.lastTimestamp() + 60000) <= System.currentTimeMillis()) {
                                    client = queue.poll();

                                    if(client == null) {
                                        break;
                                    }

                                    if(peekClient != client && (client.lastTimestamp() + 60000) > System.currentTimeMillis() ) {
                                        try {
                                            if(!queue.offerFirst(client)) {
                                                client.checkClose();
                                            }
                                        } catch (Throwable t) {
                                            client.checkClose();
                                        } finally {
                                            client = null;
                                        }
                                        continue;
                                    }

                                    try {
                                        client.send(8, heartbeatBytes);
                                        client.receive(8, 10000);
                                        Logger.info("[%d]心跳！", client.hashCode());
                                    } catch (Exception t) {
                                        Logger.error("心跳发送失败", t);
                                        client.checkClose();
                                        client = null;
                                    } finally {
                                        try {
                                            if (client != null && !queue.offer(client)) {
                                                client.checkClose();
                                            }
                                        } catch (Throwable t) {
                                            Logger.warn("心跳检查出现异常情况", t);
                                            client.checkClose();
                                        } finally {
                                            client = null;
                                        }
                                    }
                                }
                            } catch (Throwable t) {
                                Logger.warn("心跳检查出现失败", t);
                            } finally {
                                if (client != null && !queue.offer(client)) {
                                    client.checkClose();
                                }

                                if (peekClient != null && (firstConnectClient == null || firstConnectClient.lastTimestamp() > peekClient.lastTimestamp())) {
                                    firstConnectClient = peekClient;
                                    firstConnectionIdx = location;
                                }
//                                Logger.info("完成一次心跳检查和发送");
                            }
                        }
                    } catch (Throwable t) {
                        Logger.warn("idle heartbeat exception", t);
                    }

                    if (firstConnectClient == null) {
                        ThreadTools.sleep(60000);
                    } else {
                        ThreadTools.sleep(60000 + firstConnectClient.lastTimestamp() - System.currentTimeMillis());
                    }
                }
            } finally {
                taskList.clear();
            }
        }, "Client-Heartbeat");
        thread.setDaemon(true);
        thread.start();
    }

    private final Lock mainLock = new ReentrantLock();
    private AtomicInteger connectCountCtrl = new AtomicInteger(0);
    private int maxConnectCount;
    private Integer connTimes;
    private Integer idleTimeout;
    private BlockingQueue<ConnectClient> connQueue;
    private String ip;
    private int port;
    private Charset charset;
    private int preLength;

    public ClientLongConnectionPool() {
    }

    public static ClientLongConnectionPool build(int maxConnectCount, int connTimes, int idleTimeout, String ip, int port,
                                                 int preLength, String charset) {
        ClientLongConnectionPool connectionPool = new ClientLongConnectionPool();
        connectionPool.setMaxConnectCount(maxConnectCount);
        connectionPool.setIdleTimeout(idleTimeout);
        connectionPool.setIp(ip);
        connectionPool.setPort(port);
        connectionPool.setPreLength(preLength);
        connectionPool.setConnTimes(connTimes);
        connectionPool.setCharset(Charset.forName(charset));
        connectionPool.setConnQueue(new LinkedBlockingDeque<>(maxConnectCount));
        addIdleTask(connectionPool);
        return connectionPool;
    }

    public static synchronized void addIdleTask(ClientLongConnectionPool connectionPool) {
        taskList.add((BlockingDeque<ConnectClient>) connectionPool.getConnQueue());
    }

    public String sendAndReceive(String request) throws Exception {
        ConnectClient socketClient = null;
        try {
            socketClient = getSocketClient();
            socketClient.send(preLength, request.getBytes(charset));
            byte[] receive = socketClient.receive(preLength, idleTimeout);
            return new String(receive, charset);
        } catch (Throwable t) {
            if (socketClient != null) {
                socketClient.checkClose();
            }
            throw t;
        } finally {
            if (socketClient != null) {
//                Logger.info("使用了网络连接: [%s]-[%s]", socketClient.getName(), socketClient.socketHash());
                insureReturnClient(socketClient);
            }
        }
    }

    private void insureReturnClient(ConnectClient socketClient) {
        if (socketClient == null) {
            return;
        }
        try {
            if (!connQueue.offer(socketClient)) {
                Logger.warn(" return false");
                socketClient.checkClose();
            }
        } catch (Throwable t) {
            socketClient.checkClose();
            Logger.warn("", t);
        }
    }

    /* 不加锁的方式获取连接和初始化连接 */
    private ConnectClient getSocketClient() throws InterruptedException {
        int exceptCount = 0;
        while ((exceptCount = connectCountCtrl.get()) < maxConnectCount) {
            ConnectClient connectClient = connQueue.poll();
            if (connectClient != null) {
                return connectClient;
            }

            if (connQueue.isEmpty() && (exceptCount = connectCountCtrl.get()) < maxConnectCount) {
                connectClient = addConnectClient(exceptCount);
                if (connectClient != null) {
                    return connectClient;
                }
            }
        }

        ConnectClient connectClient = connQueue.poll(connTimes, TimeUnit.SECONDS);
        if (connectClient == null) {
            throw new RuntimeException("获取连接超时");
        }
        return connectClient;
    }

    /* 尝试创建一个新连接 */
    private ConnectClient addConnectClient(int exceptCount) {
        if (exceptCount < maxConnectCount && connectCountCtrl.compareAndSet(exceptCount, exceptCount + 1)) {
            return ConnectClient.build(ip, port, connTimes, idleTimeout);
        }
        return null;
    }

    public Lock getMainLock() {
        return mainLock;
    }

    public int getMaxConnectCount() {
        return maxConnectCount;
    }

    public void setMaxConnectCount(int maxConnectCount) {
        this.maxConnectCount = maxConnectCount;
    }

    public AtomicInteger getConnectCountCtrl() {
        return connectCountCtrl;
    }

    public void setConnectCountCtrl(AtomicInteger connectCountCtrl) {
        this.connectCountCtrl = connectCountCtrl;
    }

    public Integer getConnTimes() {
        return connTimes;
    }

    public void setConnTimes(Integer connTimes) {
        this.connTimes = connTimes;
    }

    public Integer getIdleTimeout() {
        return idleTimeout;
    }

    public void setIdleTimeout(Integer idleTimeout) {
        this.idleTimeout = idleTimeout;
    }

    public BlockingQueue<ConnectClient> getConnQueue() {
        return connQueue;
    }

    public void setConnQueue(BlockingQueue<ConnectClient> connQueue) {
        this.connQueue = connQueue;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public Charset getCharset() {
        return charset;
    }

    public void setCharset(Charset charset) {
        this.charset = charset;
    }

    public int getPreLength() {
        return preLength;
    }

    public void setPreLength(int preLength) {
        this.preLength = preLength;
    }
}
